package com.zhengb.flink.study.sql.streamsql;

import com.alibaba.fastjson.JSONObject;
import com.zhengb.flink.study.sql.datasource.DriverLocationSource;
import com.zhengb.flink.study.sql.datasource.LocationEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;

/**
 * @author zhengbo
 * @date 2020/1/2
 */
@Slf4j
public class DriverLocationStreamSql {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        DataStreamSource<LocationEntity> dataStreamSource = env.addSource(new DriverLocationSource());

        tableEnvironment.registerDataStream("driverLocation", dataStreamSource, "driverNo,lon,lat");

        Table table = tableEnvironment.sqlQuery("select * from driverLocation where driverNo='420356'");

        tableEnvironment.toAppendStream(table, LocationEntity.class).addSink(new SinkFunction<LocationEntity>() {
            @Override
            public void invoke(LocationEntity value, Context context) throws Exception {
                log.info("sink driver source data:{}", JSONObject.toJSONString(value));
            }
        });


        env.execute("DriverLocationStreamSql");
    }
}
